package com.zzx.graduation.controller;

import com.zzx.graduation.utils.KafkaBatchConsumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

@Component
@ServerEndpoint("/groupWebSocket")
@Slf4j
public class GroupWebSocket {

    private Session session;
    //如果消息发往群聊，则消息要发到kafka,主题tpoic为：group--targetid(群id)
    //发往个人主题邮箱topic： mail--id （代表这个id的人的收件箱，存放收到的消息）
    private Integer groupId;
    private Integer userId;
    private volatile KafkaBatchConsumer kafkaBatchConsumer;
    private Thread thread = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("进入消费线程准备消费------------>>>>>>>>");
            //  测试消费
            kafkaBatchConsumer = new KafkaBatchConsumer("group--" + groupId, userId + "-") {
                @Override
                public void hanldeRecord(ConsumerRecord<String, String> record) {
                    String str = record.value();
                    System.out.println("推送群聊消息--->" + str );
                    //消息推送
                    try {
                        session.getBasicRemote().sendText(str);
                        System.out.println("群聊消息消息推送成功");
                    } catch (Exception e) {
                        System.out.println("群聊消息消息推送失败");
                        e.printStackTrace();
                    }
                }
            };
            kafkaBatchConsumer.receivebatch();
        }
    });

    private static CopyOnWriteArraySet<GroupWebSocket> webSocketSet = new CopyOnWriteArraySet<>();
     private static ConcurrentHashMap<Integer, GroupWebSocket> webSocketMap = new ConcurrentHashMap<>();


    @OnOpen
    public void onOpen(Session session) {
        //userId=? & groupId = ?
        this.session = session;
        System.out.println(session.getQueryString());
        String[] split = session.getQueryString().split("&");

        String[] userParam = split[0].split("=");
        String userIdKey = userParam[0];
        Integer userIdValue = Integer.valueOf(userParam[1]);

        String[] groupParam = split[1].split("=");
        String groupIdKey = groupParam[0];
        Integer groupIdValue = Integer.valueOf(groupParam[1]);

        if ("groupId".equals(groupIdKey) && "userId".equals(userIdKey) ) {
            this.groupId = groupIdValue;
            this.userId = userIdValue;
            GroupWebSocket orDefault = webSocketMap.getOrDefault(this.userId, null);
            int id = this.userId;
            if (orDefault != null) {
                //说明已经存在
                final GroupWebSocket[] isEx = new GroupWebSocket[1];
                webSocketSet.forEach(
                        e -> {
                          if (e.userId == id) {
                           isEx[0] = e;
                              //关闭kafka消费者
                          }
                        }
                );
                if (isEx[0].kafkaBatchConsumer != null) {
                    isEx[0].kafkaBatchConsumer.flag = false;
                    webSocketSet.remove(isEx[0]);
                }

           }
            webSocketSet.add(this);
            webSocketMap.put(userIdValue, this);
            log.info("【groupwebsocket消息】有新的连接, 总数:{}", webSocketSet.size());
            //确定登录状态，开启消费者，开始接受消息，拉取kafka
            this.thread.start();
        }else {
            try {
                session.getBasicRemote().sendText("group:没有传入id或者 此id已经在线,正在处理中");
            } catch (IOException e) {
                e.printStackTrace();
            }finally {
                onClose();
            }
        }

    }

    @OnClose
    public void onClose() {
        webSocketSet.remove(this);
        webSocketMap.remove(this.userId);
        //关闭kafka消费者
        this.kafkaBatchConsumer.flag = false;
        log.info("【groupwebsocket消息】连接断开, 总数:{}", webSocketSet.size());
    }

    @OnMessage
    public void onMessage(String message) {
        log.info("【groupwebsocket消息】收到客户端发来的消息:{}", message);
    }

    public void sendMessage(String message) {
        for (GroupWebSocket webSocket : webSocketSet) {
            log.info("【groupwebsocket消息】广播消息, message={}", message);
            try {
                webSocket.session.getBasicRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}
